//! Database introspection service for analyzing database structure
//! 
//! This module provides functionality to introspect database schemas and
//! extract metadata about tables, columns, and relationships.

use crate::{
    models::{
        ColumnInfo, DatabaseMetadata, ForeignKeyInfo, RelationshipInfo, RelationshipType,
        TableInfo,
    },
    Error, Result,
};
use sea_orm::{ConnectionTrait, DatabaseConnection, Statement};
use std::sync::Arc;
use std::collections::HashMap;
use tracing::{debug, info, warn};

/// Database introspector for analyzing database structure
pub struct DatabaseIntrospector {
    connection: Arc<DatabaseConnection>,
}

impl DatabaseIntrospector {
    /// Create a new database introspector
    pub fn new(connection: Arc<DatabaseConnection>) -> Self {
        Self { connection }
    }
    
    /// Get complete database metadata
    pub async fn get_database_metadata(&self) -> Result<DatabaseMetadata> {
        info!("Starting database introspection");
        
        let database_type = match self.connection.get_database_backend() {
            sea_orm::DatabaseBackend::Postgres => "PostgreSQL".to_string(),
            sea_orm::DatabaseBackend::MySql => "MySQL".to_string(),
            sea_orm::DatabaseBackend::Sqlite => "SQLite".to_string(),
        };
        
        let mut metadata = DatabaseMetadata::new(database_type);
        
        // Get all tables
        let tables = self.get_tables().await?;
        info!("Found {} tables", tables.len());
        
        for mut table in tables {
            debug!("Processing table: {}", table.name);
            
            // Get columns for this table
            let columns = self.get_columns(&table.name).await?;
            for column in columns {
                table.add_column(column);
            }
            
            // Get foreign keys for this table
            let foreign_keys = self.get_foreign_keys(&table.name).await?;
            for fk in foreign_keys {
                table.add_foreign_key(fk);
            }
            
            metadata.add_table(table);
        }
        
        // Get relationships
        let relationships = self.get_relationships().await?;
        for relationship in relationships {
            metadata.add_relationship(relationship);
        }
        
        info!("Database introspection completed successfully");
        Ok(metadata)
    }
    
    /// Get all tables in the database
    pub async fn get_tables(&self) -> Result<Vec<TableInfo>> {
        let query = match self.connection.get_database_backend() {
            sea_orm::DatabaseBackend::Postgres => {
                r#"
                SELECT 
                    table_name,
                    table_schema,
                    COALESCE(obj_description(c.oid), '') as table_comment
                FROM information_schema.tables t
                LEFT JOIN pg_class c ON c.relname = t.table_name
                LEFT JOIN pg_namespace n ON n.oid = c.relnamespace AND n.nspname = t.table_schema
                WHERE table_type = 'BASE TABLE'
                AND table_schema NOT IN ('information_schema', 'pg_catalog', 'pg_toast')
                ORDER BY table_schema, table_name
                "#
            }
            sea_orm::DatabaseBackend::MySql => {
                r#"
                SELECT 
                    table_name,
                    table_schema,
                    table_comment
                FROM information_schema.tables
                WHERE table_type = 'BASE TABLE'
                AND table_schema NOT IN ('information_schema', 'performance_schema', 'mysql', 'sys')
                ORDER BY table_schema, table_name
                "#
            }
            sea_orm::DatabaseBackend::Sqlite => {
                r#"
                SELECT 
                    name as table_name,
                    'main' as table_schema,
                    '' as table_comment
                FROM sqlite_master
                WHERE type = 'table'
                AND name NOT LIKE 'sqlite_%'
                ORDER BY name
                "#
            }
        };
        
        let rows = self.connection
            .query_all(Statement::from_string(
                self.connection.get_database_backend(),
                query.to_string(),
            ))
            .await?;
        
        let mut tables = Vec::new();
        for row in rows {
            let table_name: String = row.try_get("", "table_name")?;
            let table_schema: String = row.try_get("", "table_schema")?;
            let table_comment: Option<String> = row.try_get("", "table_comment").ok();
            
            let mut table = TableInfo::new(table_name, table_schema);
            table.comment = table_comment.filter(|c| !c.is_empty());
            tables.push(table);
        }
        
        Ok(tables)
    }
    
    /// Get columns for a specific table
    pub async fn get_columns(&self, table_name: &str) -> Result<Vec<ColumnInfo>> {
        let query = match self.connection.get_database_backend() {
            sea_orm::DatabaseBackend::Postgres => {
                r#"
                SELECT 
                    c.column_name,
                    c.data_type,
                    c.is_nullable,
                    c.column_default,
                    c.character_maximum_length,
                    c.numeric_precision,
                    c.numeric_scale,
                    COALESCE(col_description(pgc.oid, c.ordinal_position), '') as column_comment,
                    CASE WHEN pk.column_name IS NOT NULL THEN true ELSE false END as is_primary_key,
                    CASE WHEN c.column_default LIKE 'nextval%' THEN true ELSE false END as is_auto_increment
                FROM information_schema.columns c
                LEFT JOIN pg_class pgc ON pgc.relname = c.table_name
                LEFT JOIN pg_namespace pgn ON pgn.oid = pgc.relnamespace AND pgn.nspname = c.table_schema
                LEFT JOIN (
                    SELECT ku.column_name
                    FROM information_schema.table_constraints tc
                    JOIN information_schema.key_column_usage ku ON tc.constraint_name = ku.constraint_name
                    WHERE tc.constraint_type = 'PRIMARY KEY' AND tc.table_name = $1
                ) pk ON pk.column_name = c.column_name
                WHERE c.table_name = $1
                ORDER BY c.ordinal_position
                "#
            }
            sea_orm::DatabaseBackend::MySql => {
                r#"
                SELECT 
                    column_name,
                    data_type,
                    is_nullable,
                    column_default,
                    character_maximum_length,
                    numeric_precision,
                    numeric_scale,
                    column_comment,
                    CASE WHEN column_key = 'PRI' THEN true ELSE false END as is_primary_key,
                    CASE WHEN extra = 'auto_increment' THEN true ELSE false END as is_auto_increment
                FROM information_schema.columns
                WHERE table_name = ?
                ORDER BY ordinal_position
                "#
            }
            sea_orm::DatabaseBackend::Sqlite => {
                format!(r#"PRAGMA table_info({})"#, table_name).leak()
            }
        };
        
        let rows = if matches!(self.connection.get_database_backend(), sea_orm::DatabaseBackend::Sqlite) {
            self.connection
                .query_all(Statement::from_string(
                    self.connection.get_database_backend(),
                    query,
                ))
                .await?
        } else {
            self.connection
                .query_all(Statement::from_sql_and_values(
                    self.connection.get_database_backend(),
                    query,
                    vec![table_name.into()],
                ))
                .await?
        };
        
        let mut columns = Vec::new();
        for row in rows {
            let column = if matches!(self.connection.get_database_backend(), sea_orm::DatabaseBackend::Sqlite) {
                // SQLite PRAGMA table_info returns different column names
                let name: String = row.try_get("", "name")?;
                let data_type: String = row.try_get("", "type")?;
                let not_null: i32 = row.try_get("", "notnull")?;
                let default_value: Option<String> = row.try_get("", "dflt_value").ok();
                let is_pk: i32 = row.try_get("", "pk")?;
                
                ColumnInfo {
                    name,
                    data_type,
                    is_nullable: not_null == 0,
                    is_primary_key: is_pk > 0,
                    is_auto_increment: false, // SQLite doesn't have explicit auto_increment
                    default_value,
                    comment: None,
                    max_length: None,
                    precision: None,
                    scale: None,
                }
            } else {
                let name: String = row.try_get("", "column_name")?;
                let data_type: String = row.try_get("", "data_type")?;
                let is_nullable_str: String = row.try_get("", "is_nullable")?;
                let is_nullable = is_nullable_str.to_uppercase() == "YES";
                let default_value: Option<String> = row.try_get("", "column_default").ok();
                let max_length: Option<i32> = row.try_get("", "character_maximum_length").ok();
                let precision: Option<i32> = row.try_get("", "numeric_precision").ok();
                let scale: Option<i32> = row.try_get("", "numeric_scale").ok();
                let comment: Option<String> = row.try_get("", "column_comment").ok();
                let is_primary_key: bool = row.try_get("", "is_primary_key").unwrap_or(false);
                let is_auto_increment: bool = row.try_get("", "is_auto_increment").unwrap_or(false);
                
                ColumnInfo {
                    name,
                    data_type,
                    is_nullable,
                    is_primary_key,
                    is_auto_increment,
                    default_value: default_value.filter(|v| !v.is_empty()),
                    comment: comment.filter(|c| !c.is_empty()),
                    max_length,
                    precision,
                    scale,
                }
            };
            
            columns.push(column);
        }
        
        Ok(columns)
    }
    
    /// Get columns for a specific table within a specific schema (real-time, schema-aware)
    pub async fn get_columns_in_schema(&self, table_schema: &str, table_name: &str) -> Result<Vec<ColumnInfo>> {
        let query = match self.connection.get_database_backend() {
            sea_orm::DatabaseBackend::Postgres => {
                r#"
                SELECT 
                    c.column_name,
                    c.data_type,
                    c.is_nullable,
                    c.column_default,
                    c.character_maximum_length,
                    c.numeric_precision,
                    c.numeric_scale,
                    COALESCE(col_description(pgc.oid, c.ordinal_position), '') as column_comment,
                    CASE WHEN pk.column_name IS NOT NULL THEN true ELSE false END as is_primary_key,
                    CASE WHEN c.column_default LIKE 'nextval%' THEN true ELSE false END as is_auto_increment
                FROM information_schema.columns c
                LEFT JOIN pg_class pgc ON pgc.relname = c.table_name
                LEFT JOIN pg_namespace pgn ON pgn.oid = pgc.relnamespace AND pgn.nspname = c.table_schema
                LEFT JOIN (
                    SELECT ku.column_name
                    FROM information_schema.table_constraints tc
                    JOIN information_schema.key_column_usage ku ON tc.constraint_name = ku.constraint_name
                    WHERE tc.constraint_type = 'PRIMARY KEY' AND tc.table_name = $1 AND tc.table_schema = $2
                ) pk ON pk.column_name = c.column_name
                WHERE c.table_name = $1 AND c.table_schema = $2
                ORDER BY c.ordinal_position
                "#
            }
            sea_orm::DatabaseBackend::MySql => {
                r#"
                SELECT 
                    column_name,
                    data_type,
                    is_nullable,
                    column_default,
                    character_maximum_length,
                    numeric_precision,
                    numeric_scale,
                    column_comment,
                    CASE WHEN column_key = 'PRI' THEN true ELSE false END as is_primary_key,
                    CASE WHEN extra = 'auto_increment' THEN true ELSE false END as is_auto_increment
                FROM information_schema.columns
                WHERE table_schema = ? AND table_name = ?
                ORDER BY ordinal_position
                "#
            }
            sea_orm::DatabaseBackend::Sqlite => {
                // SQLite 无 schema 概念，沿用 PRAGMA
                format!(r#"PRAGMA table_info({})"#, table_name).leak()
            }
        };

        let rows = if matches!(self.connection.get_database_backend(), sea_orm::DatabaseBackend::Sqlite) {
            self.connection
                .query_all(Statement::from_string(
                    self.connection.get_database_backend(),
                    query,
                ))
                .await?
        } else {
            // 注意参数顺序：Postgres 为 (table_name, table_schema)，MySQL 为 (table_schema, table_name)
            match self.connection.get_database_backend() {
                sea_orm::DatabaseBackend::Postgres => {
                    self.connection
                        .query_all(Statement::from_sql_and_values(
                            self.connection.get_database_backend(),
                            query,
                            vec![table_name.into(), table_schema.into()],
                        ))
                        .await?
                }
                sea_orm::DatabaseBackend::MySql => {
                    self.connection
                        .query_all(Statement::from_sql_and_values(
                            self.connection.get_database_backend(),
                            query,
                            vec![table_schema.into(), table_name.into()],
                        ))
                        .await?
                }
                sea_orm::DatabaseBackend::Sqlite => unreachable!(),
            }
        };

        let mut columns = Vec::new();
        for row in rows {
            let column = if matches!(self.connection.get_database_backend(), sea_orm::DatabaseBackend::Sqlite) {
                let name: String = row.try_get("", "name")?;
                let data_type: String = row.try_get("", "type")?;
                let not_null: i32 = row.try_get("", "notnull")?;
                let default_value: Option<String> = row.try_get("", "dflt_value").ok();
                let is_pk: i32 = row.try_get("", "pk")?;
                
                ColumnInfo {
                    name,
                    data_type,
                    is_nullable: not_null == 0,
                    is_primary_key: is_pk > 0,
                    is_auto_increment: false,
                    default_value,
                    comment: None,
                    max_length: None,
                    precision: None,
                    scale: None,
                }
            } else {
                let name: String = row.try_get("", "column_name")?;
                let data_type: String = row.try_get("", "data_type")?;
                let is_nullable_str: String = row.try_get("", "is_nullable")?;
                let is_nullable = is_nullable_str.to_uppercase() == "YES";
                let default_value: Option<String> = row.try_get("", "column_default").ok();
                let max_length: Option<i32> = row.try_get("", "character_maximum_length").ok();
                let precision: Option<i32> = row.try_get("", "numeric_precision").ok();
                let scale: Option<i32> = row.try_get("", "numeric_scale").ok();
                let comment: Option<String> = row.try_get("", "column_comment").ok();
                let is_primary_key: bool = row.try_get("", "is_primary_key").unwrap_or(false);
                let is_auto_increment: bool = row.try_get("", "is_auto_increment").unwrap_or(false);
                
                ColumnInfo {
                    name,
                    data_type,
                    is_nullable,
                    is_primary_key,
                    is_auto_increment,
                    default_value: default_value.filter(|v| !v.is_empty()),
                    comment: comment.filter(|c| !c.is_empty()),
                    max_length,
                    precision,
                    scale,
                }
            };
            columns.push(column);
        }
        Ok(columns)
    }
    
    /// Get foreign keys for a specific table
    pub async fn get_foreign_keys(&self, table_name: &str) -> Result<Vec<ForeignKeyInfo>> {
        let query = match self.connection.get_database_backend() {
            sea_orm::DatabaseBackend::Postgres => {
                r#"
                SELECT 
                    tc.constraint_name,
                    tc.table_name as from_table,
                    kcu.column_name as from_column,
                    ccu.table_name as to_table,
                    ccu.column_name as to_column,
                    rc.update_rule as on_update,
                    rc.delete_rule as on_delete
                FROM information_schema.table_constraints tc
                JOIN information_schema.key_column_usage kcu ON tc.constraint_name = kcu.constraint_name
                JOIN information_schema.constraint_column_usage ccu ON ccu.constraint_name = tc.constraint_name
                JOIN information_schema.referential_constraints rc ON rc.constraint_name = tc.constraint_name
                WHERE tc.constraint_type = 'FOREIGN KEY' AND tc.table_name = $1
                "#
            }
            sea_orm::DatabaseBackend::MySql => {
                r#"
                SELECT 
                    kcu.constraint_name,
                    kcu.table_name as from_table,
                    kcu.column_name as from_column,
                    kcu.referenced_table_name as to_table,
                    kcu.referenced_column_name as to_column,
                    rc.update_rule as on_update,
                    rc.delete_rule as on_delete
                FROM information_schema.key_column_usage kcu
                JOIN information_schema.referential_constraints rc ON kcu.constraint_name = rc.constraint_name
                WHERE kcu.referenced_table_name IS NOT NULL AND kcu.table_name = ?
                "#
            }
            sea_orm::DatabaseBackend::Sqlite => {
                format!(r#"PRAGMA foreign_key_list({})"#, table_name).leak()
            }
        };
        
        let rows = if matches!(self.connection.get_database_backend(), sea_orm::DatabaseBackend::Sqlite) {
            self.connection
                .query_all(Statement::from_string(
                    self.connection.get_database_backend(),
                    query,
                ))
                .await?
        } else {
            self.connection
                .query_all(Statement::from_sql_and_values(
                    self.connection.get_database_backend(),
                    query,
                    vec![table_name.into()],
                ))
                .await?
        };
        
        let mut foreign_keys = Vec::new();
        for row in rows {
            let fk = if matches!(self.connection.get_database_backend(), sea_orm::DatabaseBackend::Sqlite) {
                // SQLite PRAGMA foreign_key_list returns different column names
                let id: i32 = row.try_get("", "id")?;
                let table: String = row.try_get("", "table")?;
                let from: String = row.try_get("", "from")?;
                let to: String = row.try_get("", "to")?;
                let on_update: String = row.try_get("", "on_update")?;
                let on_delete: String = row.try_get("", "on_delete")?;
                
                ForeignKeyInfo {
                    constraint_name: format!("fk_{}_{}", table_name, id),
                    from_table: table_name.to_string(),
                    from_column: from,
                    to_table: table,
                    to_column: to,
                    on_update: Some(on_update),
                    on_delete: Some(on_delete),
                }
            } else {
                let constraint_name: String = row.try_get("", "constraint_name")?;
                let from_table: String = row.try_get("", "from_table")?;
                let from_column: String = row.try_get("", "from_column")?;
                let to_table: String = row.try_get("", "to_table")?;
                let to_column: String = row.try_get("", "to_column")?;
                let on_update: Option<String> = row.try_get("", "on_update").ok();
                let on_delete: Option<String> = row.try_get("", "on_delete").ok();
                
                ForeignKeyInfo {
                    constraint_name,
                    from_table,
                    from_column,
                    to_table,
                    to_column,
                    on_update,
                    on_delete,
                }
            };
            
            foreign_keys.push(fk);
        }
        
        Ok(foreign_keys)
    }
    
    /// Get relationships between tables
    pub async fn get_relationships(&self) -> Result<Vec<RelationshipInfo>> {
        let foreign_keys = self.get_all_foreign_keys().await?;
        let mut relationships = Vec::new();
        
        for fk in foreign_keys {
            // Determine relationship type based on foreign key constraints
            let relationship_type = self.determine_relationship_type(&fk).await?;
            
            let relationship = RelationshipInfo {
                relationship_type,
                from_table: fk.from_table,
                from_field: fk.from_column,
                to_table: fk.to_table,
                to_field: fk.to_column,
                junction_table: None, // TODO: Detect many-to-many relationships
            };
            
            relationships.push(relationship);
        }
        
        Ok(relationships)
    }
    
    /// Get all foreign keys in the database
    async fn get_all_foreign_keys(&self) -> Result<Vec<ForeignKeyInfo>> {
        let tables = self.get_tables().await?;
        let mut all_foreign_keys = Vec::new();
        
        for table in tables {
            let foreign_keys = self.get_foreign_keys(&table.name).await?;
            all_foreign_keys.extend(foreign_keys);
        }
        
        Ok(all_foreign_keys)
    }
    
    /// Determine the type of relationship based on foreign key
    async fn determine_relationship_type(&self, fk: &ForeignKeyInfo) -> Result<RelationshipType> {
        // For now, assume all foreign keys represent many-to-one relationships
        // TODO: Implement more sophisticated relationship detection
        Ok(RelationshipType::ManyToOne)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use sea_orm::{MockDatabase, MockRow};
    
    #[tokio::test]
    async fn test_introspector_creation() {
        let db = MockDatabase::new(sea_orm::DatabaseBackend::Postgres)
            .into_connection();
        
        let introspector = DatabaseIntrospector::new(Arc::new(db));
        assert!(true);
    }
    
    #[tokio::test]
    async fn test_empty_database_metadata() {
        let db = MockDatabase::new(sea_orm::DatabaseBackend::Postgres)
            .append_query_results(vec![
                Vec::<MockRow>::new(), // Empty tables result
            ])
            .into_connection();
        
        let introspector = DatabaseIntrospector::new(Arc::new(db));
        let metadata = introspector.get_database_metadata().await.unwrap();
        
        assert_eq!(metadata.database_type, "PostgreSQL");
        assert!(metadata.tables.is_empty());
    }
}